package org.example.flink.source;

import java.util.Random;
import java.util.concurrent.TimeUnit;

import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import scala.Tuple3;

public class S1 implements SourceFunction<Tuple3<String, Integer, Integer>> {

    private static Logger s1 = LoggerFactory.getLogger("s1");

    private boolean flag = true;
    static final String[] NAMES = {"tom", "jerry", "alice", "bob", "john", "grace"};
    private static final Random rnd = new Random();

    @Override
    public void run(SourceContext<Tuple3<String, Integer, Integer>> sourceContext) throws Exception {
        int cnt = 0;
        // while (flag){
        // Tuple3 t = new Tuple3(NAMES[rnd.nextInt(6)],rnd.nextInt(10),++cnt);
        // sourceContext.collect(t);
        // s1.info(t.toString());
        // TimeUnit.MILLISECONDS.sleep(500);
        // }
        int num = 20;
        for (int i = 0; i < num; i++) {
            Tuple3 t = new Tuple3(NAMES[rnd.nextInt(6)], rnd.nextInt(10), ++cnt);
            sourceContext.collect(t);
            s1.info(t.toString());
            TimeUnit.MILLISECONDS.sleep(500);
        }
        System.out.println("s1-------end");

    }

    @Override
    public void cancel() {
        flag = false;
    }

}
